package com.tandashi.springbootstudy01.sensor.receiver;

import com.fasterxml.jackson.databind.JavaType;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.common.utils.Bytes;
import org.springframework.beans.factory.annotation.Autowired;

import java.io.IOException;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;

@SuppressWarnings("WeakerAccess")
@Slf4j
public abstract class JsonReceiver<T> implements Receiver {
    protected JsonReceiver() {
        Type superclass = getClass().getGenericSuperclass();
        if (superclass instanceof ParameterizedType) {
            type = ((ParameterizedType) superclass).getActualTypeArguments()[0];
        } else {
            type = null;
        }
    }


    @Override
    public void receive(Bytes message) {
        try {
            T bean = convert(message);
            if (!validate(bean)) {
                return;
            }

            beforeReceive(bean);
            doReceive(bean);
        } catch (Exception e) {
            log.error("consume sensor message error", e);
        }
    }

    protected T convert(Bytes message) throws IOException {
        JavaType javaType = objectMapper.constructType(type);
        return objectMapper.readValue(message.get(), javaType);
    }

    protected abstract boolean validate(T message);

    protected abstract void doReceive(T message) throws Exception;

    @SuppressWarnings("unused")
    protected void beforeReceive(T message) {

    }

    protected final Type type;

    @Autowired
    protected ObjectMapper objectMapper;
}
